import concurrent.futures
import datetime
import json
import os
import re
import traceback
import warnings

import numpy as np
import pandas as pd
import parsel
from loguru import logger
from retrying import retry
from tqdm import tqdm

from components.config import WFQ_SOURCE_MYSQL_CONFIG
from utils.db.mysqldb import MysqlDB

warnings.filterwarnings("ignore")


class BaseParser:
    def _parse_mall_page(self, df_one_mall):
        mall_id = df_one_mall["mall_id"].iloc[0]
        version = df_one_mall["version"].iloc[0]
        is_strict_all = df_one_mall["is_strict_all"].iloc[0]
        page_goods_count = df_one_mall["page_goods_count"].iloc[0]
        update_at = df_one_mall["create_time"].iloc[0]
        df_one_mall = df_one_mall.sort_values(by=["id"], ascending=False)
        shop_name = None
        mall_star = None
        mall_details_label_list = None
        open_time_label = None
        mall_desc = None
        mall_logo = None
        mall_label_volist = None
        mall_rating_text_list = None
        service_score = None
        shop_status = None
        flag = None
        shop_url = f"https://mobile.yangkeduo.com/mall_page.html?mall_id={mall_id}"
        for _, row in df_one_mall.iterrows():
            data = row["response"]
            if "data" in data:
                if data["data"]:
                    if "该店铺已失效" in data["html"]:
                        flag = -1
                        shop_status = "该店铺已失效"
                    mall_detail_info = data["data"]["store"].get("mallDetailInfo")
                    if mall_detail_info:
                        mall_basic_info = mall_detail_info.get("mallBasicInfo")
                        if mall_basic_info:
                            shop_name = shop_name or mall_basic_info["mallName"]
                            mall_desc = mall_desc or mall_basic_info["mallDesc"]
                            mall_logo = mall_logo or mall_basic_info["mallLogo"]
                        mall_star = mall_star or (mall_detail_info.get("dsr") or {}).get("mallStar")
                        mall_details_label_list = mall_details_label_list or mall_detail_info.get(
                            "mallDetailsLabelList"
                        )
                        mall_label_volist = mall_label_volist or mall_detail_info.get("mallLabelVolist")
                        for k in mall_label_volist or []:
                            if "年老店" in k["desc"]:
                                open_time_label = open_time_label or k["desc"]
                        mall_rating_text_list = mall_rating_text_list or (mall_detail_info.get("dsr") or {}).get(
                            "mallRatingTextList"
                        )
                        service_score = service_score or (mall_detail_info.get("dsr") or {}).get("serviceScore")
                        open_time_label = (
                            open_time_label or (mall_detail_info.get("openTimeLabel") or {"desc": None})["desc"]
                        )
                        if mall_basic_info and mall_basic_info.get("isOpen") == 0:
                            shop_status = "该店铺已失效"
                            flag = -1
                    else:
                        mall_basic_info = data["data"]["store"].get("mallBasicInfo")
                        if mall_basic_info:
                            shop_name = shop_name or mall_basic_info.get("mallName")
                            mall_desc = mall_desc or mall_basic_info.get("mallDesc")
                            mall_logo = mall_logo or mall_basic_info.get("mallLogo")
                            mall_label_volist = mall_label_volist or mall_basic_info.get("mallLabelList")
                            for k in mall_label_volist or []:
                                if "年老店" in k["desc"]:
                                    open_time_label = open_time_label or k["desc"]

                            if mall_basic_info.get("isOpen") == 0:
                                shop_status = "该店铺已失效"
                                flag = -1
                            else:
                                if mall_basic_info.get("hasGoods") is False:
                                    no_goods_list_tips = (
                                        parsel.Selector(text=data["html"])
                                        .xpath('//div[contains(@class, "no-goods-list-tips")]/text()')
                                        .get()
                                    )
                                    if no_goods_list_tips:
                                        if no_goods_list_tips == "该店铺正在上传商品":
                                            shop_status = "该店铺正在上传商品"
                                            flag = -1
                                        if no_goods_list_tips == "该店铺已失效":
                                            shop_status = "该店铺已失效"
                                            flag = -1
        if shop_status is None:
            shop_status = "开店"
            flag = 1
        item = {
            "mall_id": mall_id,
            "version": version,
            "is_strict_all": is_strict_all,
            "page_goods_count": page_goods_count,
            "update_at": update_at,
        }
        if mall_id and (shop_name or flag is not None):
            item["shop_name"] = shop_name
            item["mall_star"] = mall_star
            item["mall_details_label_list"] = mall_details_label_list
            item["shop_url"] = shop_url
            item["open_time_label"] = open_time_label
            item["mall_desc"] = mall_desc
            item["mall_logo"] = mall_logo
            item["mall_label_volist"] = mall_label_volist
            item["mall_rating_text_list"] = mall_rating_text_list
            item["service_score"] = service_score
            if shop_status:
                item["shop_status"] = shop_status
            if flag is not None:
                item["flag"] = flag
        res_df = pd.DataFrame([item])
        return res_df

    @logger.catch
    def _parse_goods(self, df_one_mall_goods):
        if df_one_mall_goods.empty:
            return
        mall_id = df_one_mall_goods["mall_id"].iloc[0]
        version = df_one_mall_goods["version"].iloc[0]
        df_one_mall_goods_new = df_one_mall_goods[
            ~(  # 异常的数据
                (df_one_mall_goods["response"] == {})
                | (df_one_mall_goods["response"].map(lambda x: "verify_auth_token" in x))
                | (df_one_mall_goods["response"] == {"error_code": 40001})
                | (df_one_mall_goods["response"] == {"error_code": 40002})
            )
        ]
        df_one_mall_goods_new.sort_values(by=["id"], ascending=True)
        df_one_mall_goods_new = df_one_mall_goods_new.drop_duplicates(subset=["page_no"], keep="last")
        results = []
        for _, row in df_one_mall_goods_new.iterrows():
            update_at = str(row["create_time"])
            goods_list = row["response"].get("goods_list") or []
            for goods in goods_list:
                goods_id = goods.get("goods_id")
                goods_name = goods.get("goods_name")
                price = goods.get("price")
                cnt = goods.get("cnt")
                sales_tip = goods.get("sales_tip")
                sale_num_by_tips = self._deal_goods_num(sales_tip)
                sales_money_calculate = None
                price = price / 100 if price else None
                if not price:
                    price = float(goods.get("price_info")) if goods.get("price_info") else None
                if sale_num_by_tips:
                    if price:
                        sales_money_calculate = sale_num_by_tips * price
                item = {
                    "mall_id": mall_id,
                    "goods_id": goods_id,
                    "goods_name": goods_name,
                    "price": price,
                    "sale_num": cnt,
                    "sales_tip": sales_tip,
                    "sale_num_by_tips": sale_num_by_tips,
                    "sales_money_calculate": sales_money_calculate,
                    "update_at": update_at,
                    "version": version,
                    "hd_url": goods.get("hd_url"),
                    "market_price": goods.get("market_price") / 100 if goods.get("market_price") else None,
                    "tag_list": goods.get("tag_list"),
                    "normal_price": goods.get("normal_price") / 100 if goods.get("normal_price") else None,
                }
                results.append(item)
        df = pd.DataFrame(results)
        if not df.empty:
            df = df.sort_values(by=["update_at"], ascending=False)
            df = df.drop_duplicates(subset=["goods_id", "version", "mall_id"], keep="first")
        return df

    def _parse_one_mall_detail(self, df_one_mall):
        mall_id = df_one_mall["mall_id"].iloc[0]
        version = df_one_mall["version"].iloc[0]
        df_one_mall.sort_values(by=["id"], ascending=False)
        shop_name = None
        mall_star = None
        mall_details_label_list = None
        open_time_label = None
        mall_desc = None
        mall_logo = None
        mall_label_volist = None
        mall_rating_text_list = None
        service_score = None
        shop_status = None
        flag = None
        shop_url = f"https://mobile.yangkeduo.com/mall_page.html?mall_id={mall_id}"
        for _, row in df_one_mall.iterrows():
            mall_detail_info = row["response"]
            if mall_detail_info:
                mall_basic_info = mall_detail_info.get("mall_basic_info")
                if mall_basic_info:
                    shop_name = shop_name or mall_basic_info.get("mall_name")
                    mall_desc = mall_desc or mall_basic_info.get("mall_desc")
                    mall_logo = mall_logo or mall_basic_info.get("mall_logo")
                else:
                    shop_name = shop_name or mall_detail_info.get("mall_name")
                    mall_desc = mall_desc or mall_detail_info.get("mall_desc")
                    mall_logo = mall_logo or mall_detail_info.get("mall_logo")
                mall_star = mall_star or (mall_detail_info.get("dsr") or {}).get("mall_star")
                mall_details_label_list = mall_details_label_list or mall_detail_info.get("mall_details_label_list")
                mall_label_volist = mall_label_volist or mall_detail_info.get("mall_label_volist")
                for k in mall_label_volist or []:
                    if "年老店" in k["desc"]:
                        open_time_label = open_time_label or k["desc"]
                mall_rating_text_list = mall_rating_text_list or (mall_detail_info.get("dsr") or {}).get(
                    "mall_rating_text_list"
                )
                service_score = service_score or (mall_detail_info.get("dsr") or {}).get("service_score")
                open_time_label = open_time_label or (mall_detail_info.get("open_time_label") or {"desc": None})["desc"]
                if mall_basic_info and mall_basic_info.get("is_open") == 0:
                    shop_status = "该店铺已失效"
                    flag = -1

        if shop_status is None:
            shop_status = "开店"
            flag = 1

        item = {
            "mall_id": mall_id,
            "version": version,
        }
        if shop_name and mall_id:
            item["shop_name"] = shop_name
            item["mall_star"] = mall_star
            item["mall_details_label_list"] = mall_details_label_list
            item["shop_url"] = shop_url
            item["open_time_label"] = open_time_label
            item["mall_desc"] = mall_desc
            item["mall_logo"] = mall_logo
            item["mall_label_volist"] = mall_label_volist
            item["mall_rating_text_list"] = mall_rating_text_list
            item["service_score"] = service_score
            if shop_status:
                item["shop_status"] = shop_status
            if flag is not None:
                item["flag"] = flag
        return pd.DataFrame([item])

    def _parse_mall_info(self, df_one_mall):
        mall_id = df_one_mall["mall_id"].iloc[0]
        version = df_one_mall["version"].iloc[0]
        df_one_mall.sort_values(by=["id"], ascending=False)
        shop_name = None
        mall_desc = None
        shop_status = None
        flag = None
        goods_num = None
        mall_sales = None
        logo = None
        is_open = None
        for _, row in df_one_mall.iterrows():
            mall_info = row["response"]
            shop_name = mall_info.get("mall_name")
            mall_desc = mall_info.get("mall_desc")
            goods_num = mall_info.get("goods_num")
            mall_sales = mall_info.get("mall_sales") or None
            logo = mall_info.get("logo")
            is_open = mall_info.get("is_open")

        item = {
            "mall_id": mall_id,
            "version": version,
        }

        shop_url = f"https://mobile.yangkeduo.com/mall_page.html?mall_id={mall_id}"

        if shop_name and mall_id:
            item["shop_name"] = shop_name
            item["shop_url"] = shop_url
            item["mall_goods_num"] = goods_num
            item["mall_sales"] = mall_sales
            item["mall_desc"] = mall_desc
            item["mall_logo"] = logo
            if is_open == 0:
                shop_status = "该店铺已失效"
                flag = -1

            if shop_status is None:
                shop_status = "开店"
                flag = 1

            if goods_num == 0:
                item["is_all"] = 1
            if shop_status:
                item["shop_status"] = shop_status
            if flag is not None:
                item["flag"] = flag
        return pd.DataFrame([item])

    def _deal_goods_num(self, goods_num_str):
        base = 1
        if not goods_num_str:
            return None
        if "万" in goods_num_str:
            base = 10000

        if "全店" in goods_num_str:
            return None
        if "总售" in goods_num_str:
            return None
        if "元" in goods_num_str:
            return None

        num = self.reg_one(
            [
                r".*售([\d.]*)",
                r".*抢([\d.]*)",
                r"[^全店]*拼([\d.]*)",
                r".*评([\d.]*)",
                r".*购([\d.]*)",
            ],
            goods_num_str,
        )
        if num:
            return float(num) * base
        else:
            if not any(c in goods_num_str for c in ["想拼"]):
                pass

    def reg_one(self, regs, text):
        for reg in regs:
            res = re.findall(reg, text)
            if res:
                return res[0]
        return

    def _parse_one_mall_combination(self, df_one_mall):
        mall_id = df_one_mall["mall_id"].iloc[0]
        version = df_one_mall["version"].iloc[0]
        df_one_mall.sort_values(by=["id"], ascending=False)
        shop_name = None
        mall_desc = None
        mall_logo = None
        shop_status = None
        flag = None
        is_strict_all = None
        crawler_goods_num = None
        shop_url = f"https://mobile.yangkeduo.com/mall_page.html?mall_id={mall_id}"
        for _, row in df_one_mall.iterrows():
            data = row["response"]
            if "result" in data:
                mall_basic_info = data["result"].get("mall_basic_info")
                if mall_basic_info:
                    shop_name = shop_name or mall_basic_info.get("mall_name")
                    mall_desc = mall_desc or mall_basic_info.get("mall_desc")
                    mall_logo = mall_logo or mall_basic_info.get("mall_logo")
                    if mall_basic_info.get("is_open") == 0:
                        shop_status = "该店铺已失效"
                        flag = -1
                    if mall_basic_info.get("has_goods") is False:
                        crawler_goods_num = 0
                        is_strict_all = 1
            elif 'redirect_url' in data:
                shop_name = '品牌好货'
                shop_status = "开店"
                flag = 1
        item = {
            "mall_id": mall_id,
            "version": version,
        }
        if shop_name and mall_id:
            item['shop_name'] = shop_name
            item['shop_url'] = shop_url
            item['mall_desc'] = mall_desc
            item['mall_logo'] = mall_logo
            if crawler_goods_num is not None:
                item['crawler_goods_num'] = crawler_goods_num
            if is_strict_all is not None:
                item['is_strict_all'] = is_strict_all
            if shop_status:
                item["shop_status"] = shop_status
            if flag is not None:
                item["flag"] = flag
        return pd.DataFrame([item])


class SyncPddSource(BaseParser):
    def __init__(self):
        self._to_db = None
        self._local_db = None
        self._table_net_pdd_proxy_log = "net_pdd_proxy_log"
        self._table_net_pdd_shop_info = "net_pdd_shop_info"
        self._table_source_pdd_shop_info = "source_net_pdd_shop_info"
        self._table_source_pdd_goods_detail = "source_result_pdd_goods_detail"

    @property
    def to_db(self):
        if self._to_db is None:
            self._to_db = MysqlDB(
                ip=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_IP"],
                port=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_PORT"],
                db=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_DB"],
                user_name=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_USER_NAME"],
                user_pass=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_USER_PASS"],
            )
        return self._to_db

    @property
    def local_db(self):
        if self._local_db is None:
            self._local_db = MysqlDB(
                ip=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_IP"],
                port=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_PORT"],
                db=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_DB"],
                user_name=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_USER_NAME"],
                user_pass=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_USER_PASS"],
            )
        return self._local_db

    def check_all(self, df_):
        result = df_["response"].map(lambda x: "has_more" in x and x["has_more"] is False).any()
        if not result:
            result = df_["response"].map(lambda x: "flip" in x and x["flip"] == '' and x["goods_list"] == []).any()
        return result

    def re_parser_mall_page(self, x):
        if x.get("data") is False:
            html = x["html"]
            try:
                if '"mallName":"' in html:
                    mallBasicInfo_reg_list = re.findall(r"mallBasicInfo\":\s*({(?:[^{}]+(?:{.*?})[^{}]+)*?})", html)
                    if mallBasicInfo_reg_list:
                        mallBasicInfo = json.loads(mallBasicInfo_reg_list[0])
                        x["data"] = {"store": {"mallBasicInfo": mallBasicInfo}}
            except:
                pass
        return x

    def fix_re_parser_mall_page(self, df_pro):
        df_pro["response"] = df_pro["response"].map(lambda x: self.re_parser_mall_page(x))
        return df_pro

    def check_mall_page_valid(self, x):
        return "data" in x and isinstance(x["data"], dict) and (
                '"mallName":"' in x["html"]
                or "comm_mall_guarantees" in x["html"]  # or '该店铺已失效' in x['html']
        )

    def set_version(self, df_pro):
        # 根据日志获取商铺的信息,按照日期分配数据版本
        mall_page_df = df_pro[
            df_pro["base_url"].str.contains("/mall_page.html") | df_pro["base_url"].str.contains("/mall_subject.html")
        ].copy()
        # logger.info(f"开始清洗数据 \n{mall_page_df.head(5)}")
        mall_page_df = self.fix_re_parser_mall_page(mall_page_df)

        # 根据日志获取商铺的店铺统计信息
        mall_info_df = df_pro[df_pro["base_url"].str.contains("/mall/{}/info")].copy()
        result_mall_info_df = pd.DataFrame()
        if not mall_info_df.empty:
            mall_info_df = mall_info_df[mall_info_df["response"].map(lambda x: "mall_name" in x)]
            if not mall_info_df.empty:
                # 额外处理app的数据
                app_mall_info = mall_info_df[~mall_info_df["mall_id"].isin(mall_page_df["mall_id"])]
                if not app_mall_info.empty:
                    new_mall_page_df = pd.DataFrame(columns=mall_page_df.columns)
                    new_mall_page_df[["mall_id", "create_time"]] = app_mall_info[["mall_id", "create_time"]]
                    new_mall_page_df["response"] = new_mall_page_df["response"].map(
                        lambda x: {"data": {}, "html": "comm_mall_guarantees"}
                    )
                    new_mall_page_df["base_url"] = "/mall_page.html"
                    mall_page_df = pd.concat([mall_page_df, new_mall_page_df], ignore_index=True)

        # 根据日志获取商铺的combination信息
        mall_combination_df = df_pro[df_pro["base_url"].str.contains("/mall_combination")].copy()
        result_mall_combination_df = pd.DataFrame()
        if not mall_combination_df.empty:
            mall_combination_df = mall_combination_df[mall_combination_df["response"].map(lambda x: any(c in x for c in ['result', 'redirect_url']))]
            if not mall_combination_df.empty:
                # 额外处理app的数据
                app_mall_combination = mall_combination_df[~mall_combination_df["mall_id"].isin(mall_page_df[mall_page_df["response"].map(self.check_mall_page_valid)]["mall_id"])]
                if not app_mall_combination.empty:
                    new_mall_page_df = pd.DataFrame(columns=mall_page_df.columns)
                    new_mall_page_df[["mall_id", "create_time"]] = app_mall_combination[["mall_id", "create_time"]]
                    new_mall_page_df["response"] = new_mall_page_df["response"].map(
                        lambda x: {"data": {}, "html": "comm_mall_guarantees"}
                    )
                    new_mall_page_df["base_url"] = "/mall_page.html"
                    mall_page_df = pd.concat([mall_page_df, new_mall_page_df], ignore_index=True)

        if not mall_page_df.empty:
            mall_page_df = mall_page_df[mall_page_df["response"].map(self.check_mall_page_valid)]
            if not mall_page_df.empty:
                mall_page_df.sort_values(["create_time"], ascending=True, inplace=True)
                mall_page_df["version"] = mall_page_df["create_time"].map(lambda x: str(x)[:10])
                mall_page_df["new_create_time"] = mall_page_df.groupby(["mall_id", "version"])["create_time"].transform(max)
                mall_page_df = mall_page_df.drop_duplicates(subset=["mall_id", "version"], keep="first")
            else:
                return {}
        else:
            return {}

        if not mall_info_df.empty:
            # 按照上面的数据版本找到在创建时间之后下一个创建时间之前，采集的店铺扩展也分配成对应的数据版本
            # 获取商铺的数据版本列表
            result_mall_info_df = (
                mall_page_df.groupby(["mall_id"])
                .apply(lambda x: self.fill_version_mall_info(mall_info_df, x.reset_index(), result_mall_info_df))
                .reset_index(drop=True)
            )

        if not mall_combination_df.empty:
            result_mall_combination_df = (
                mall_page_df.groupby(["mall_id"])
                .apply(lambda x: self.fill_version_mall_combination(mall_combination_df, x.reset_index(), result_mall_combination_df))
                .reset_index(drop=True)
            )

        result_goods_df = pd.DataFrame()
        # 根据日志获取商铺的商品明细
        goods_df = df_pro[df_pro["base_url"].str.contains("/turing/mall/query_cat_goods")].copy()
        if not goods_df.empty:
            goods_df = goods_df[goods_df["response"].map(lambda x: x is not None and "page_no" in x)]
            goods_df = goods_df[goods_df["response"].map(lambda x: x is not None and "flip" in x)]
            if not goods_df.empty:
                result_goods_df = (
                    mall_page_df.copy()
                    .groupby(["mall_id"])
                    .apply(lambda x: self.fill_version_goods(goods_df, x.reset_index(), result_goods_df))
                    .reset_index(drop=True)
                )

        category_list_df = df_pro[df_pro["base_url"].str.contains("/turing/mall/query_mall_category_list")].copy()
        result_category_list_df = pd.DataFrame()
        if not category_list_df.empty:
            result_category_list_df = (
                mall_page_df.copy()
                .groupby(["mall_id"])
                .apply(lambda x: self.fill_version_category_list(category_list_df, x.reset_index(), result_category_list_df))
                .reset_index(drop=True)
            )

        if not result_goods_df.empty:
            is_all_mapping_df = (
                result_goods_df.groupby(["mall_id", "version"])
                .apply(self.check_all)
                .reset_index()
                .rename(columns={0: "is_strict_all"})
            )
        else:
            is_all_mapping_df = mall_page_df[["mall_id", "version"]]
            is_all_mapping_df["is_strict_all"] = None

        if not result_category_list_df.empty:
            is_all_category_list_df = (
                result_category_list_df.copy()
                .groupby(["mall_id", "version"])
                .apply(lambda x: self.fill_goods_num_category_list(x))
                .reset_index(drop=True)
            )
        else:
            is_all_category_list_df = mall_page_df[["mall_id", "version"]]
            is_all_category_list_df["page_goods_count"] = None

        # 根据日志获取商铺的店铺扩展信息
        mall_detail_info_df = df_pro[df_pro["base_url"].str.contains("/query_mall_detail_info")].copy()
        result_mall_detail_info_df = pd.DataFrame()
        if not mall_detail_info_df.empty:
            mall_detail_info_df = mall_detail_info_df[mall_detail_info_df["response"].map(lambda x: "dsr" in x)]
            if not mall_detail_info_df.empty:
                # 按照上面的数据版本找到在创建时间之后下一个创建时间之前，采集的店铺扩展也分配成对应的数据版本
                # 获取商铺的数据版本列表
                result_mall_detail_info_df = (
                    mall_page_df.groupby(["mall_id"])
                    .apply(
                        lambda x: self.fill_version_mall_detail(
                            mall_detail_info_df, x.reset_index(), result_mall_detail_info_df
                        )
                    )
                    .reset_index(drop=True)
                )

        mall_page_df = pd.merge(mall_page_df, is_all_mapping_df, how="left", on=["mall_id", "version"])
        mall_page_df = pd.merge(mall_page_df, is_all_category_list_df, how="left", on=["mall_id", "version"])
        mall_page_df = mall_page_df.replace({np.nan: None})

        return {
            "mall_page_df": mall_page_df,
            "goods_df": result_goods_df,
            "result_mall_detail_info_df": result_mall_detail_info_df,
            "result_mall_info_df": result_mall_info_df,
            "result_mall_combination_df": result_mall_combination_df,
        }

    def fill_version_category_list(self, category_list_df, mall_page_df, result_category_list_df):
        category_list_df = category_list_df[category_list_df["mall_id"].isin(mall_page_df["mall_id"])]
        if category_list_df.empty:
            return result_category_list_df
        versions = mall_page_df["version"].sort_values(ascending=True).unique()
        for version in versions:
            # 获取当前版本的最后一个创建时间
            last_create_time = mall_page_df[mall_page_df["version"] == version]["create_time"].max()
            # 允许误差
            last_create_time = str(
                datetime.datetime.strptime(last_create_time, "%Y-%m-%d %H:%M:%S") - datetime.timedelta(seconds=10)
            )
            # 获取下一个版本的第一个创建时间
            next_version_first_create_time = mall_page_df[mall_page_df["version"] > version]["create_time"].min()

            if next_version_first_create_time is np.nan:
                next_version_first_create_time = "2099-01-01"

            # 筛选出在当前版本的最后一个创建时间和下一个版本的第一个创建时间之间的商品明细
            category_list_df_in_version = category_list_df[
                (category_list_df["create_time"] >= last_create_time)
                & (category_list_df["create_time"] < next_version_first_create_time)
            ]
            category_list_df_in_version["version"] = version
            result_category_list_df = pd.concat(
                [result_category_list_df, category_list_df_in_version], ignore_index=True
            )
        return result_category_list_df

    def fill_goods_num_category_list(self, df):
        def deal_response(response_dict):
            if 'category_list' in response_dict:
                category_list = response_dict['category_list']
                for category in category_list:
                    if "全部商品" in category['name']:
                        return category['goods_count']
        if df.empty:
            return None
        df['page_goods_count'] = df["response"].map(lambda x: deal_response(x))
        df.sort_values(by=['create_time'], ascending=False, inplace=True)
        return df[['mall_id', 'version', 'page_goods_count']].drop_duplicates()

    def fill_version_mall_combination(self, mall_combination_df, mall_page_df, result_mall_combination_df):
        mall_combination_df = mall_combination_df[mall_combination_df["mall_id"].isin(mall_page_df["mall_id"])]
        if mall_combination_df.empty:
            return result_mall_combination_df
        versions = mall_page_df["version"].sort_values(ascending=True).unique()
        for version in versions:
            # 获取当前版本的最后一个创建时间
            last_create_time = mall_page_df[mall_page_df["version"] == version]["create_time"].max()
            # 允许误差
            last_create_time = str(
                datetime.datetime.strptime(last_create_time, "%Y-%m-%d %H:%M:%S") - datetime.timedelta(seconds=10)
            )
            # 获取下一个版本的第一个创建时间
            next_version_first_create_time = mall_page_df[mall_page_df["version"] > version]["create_time"].min()

            if next_version_first_create_time is np.nan:
                next_version_first_create_time = "2099-01-01"

            # 筛选出在当前版本的最后一个创建时间和下一个版本的第一个创建时间之间的商品明细
            mall_combination_df_in_version = mall_combination_df[
                (mall_combination_df["create_time"] >= last_create_time)
                & (mall_combination_df["create_time"] < next_version_first_create_time)
            ]
            mall_combination_df_in_version["version"] = version
            result_mall_combination_df = pd.concat(
                [result_mall_combination_df, mall_combination_df_in_version], ignore_index=True
            )
        return result_mall_combination_df

    def fill_version_mall_info(self, mall_info_df, mall_page_df, result_mall_info_df):
        mall_info_df = mall_info_df[mall_info_df["mall_id"].isin(mall_page_df["mall_id"])]
        if mall_info_df.empty:
            return result_mall_info_df
        versions = mall_page_df["version"].sort_values(ascending=True).unique()
        for version in versions:
            # 获取当前版本的最后一个创建时间
            last_create_time = mall_page_df[mall_page_df["version"] == version]["create_time"].max()
            # 允许误差
            last_create_time = str(
                datetime.datetime.strptime(last_create_time, "%Y-%m-%d %H:%M:%S") - datetime.timedelta(seconds=10)
            )
            # 获取下一个版本的第一个创建时间
            next_version_first_create_time = mall_page_df[mall_page_df["version"] > version]["create_time"].min()

            if next_version_first_create_time is np.nan:
                next_version_first_create_time = "2099-01-01"

            # 筛选出在当前版本的最后一个创建时间和下一个版本的第一个创建时间之间的商品明细
            mall_info_in_version = mall_info_df[
                (mall_info_df["create_time"] >= last_create_time)
                & (mall_info_df["create_time"] < next_version_first_create_time)
            ]
            mall_info_in_version["version"] = version
            result_mall_info_df = pd.concat([result_mall_info_df, mall_info_in_version], ignore_index=True)
        return result_mall_info_df

    def fill_version_mall_detail(self, mall_detail_info_df, mall_page_df, result_mall_detail_info_df):
        mall_detail_info_df = mall_detail_info_df[mall_detail_info_df["mall_id"].isin(mall_page_df["mall_id"])]
        if mall_detail_info_df.empty:
            return result_mall_detail_info_df
        versions = mall_page_df["version"].sort_values(ascending=True).unique()
        for version in versions:
            # 获取当前版本的最后一个创建时间
            last_create_time = mall_page_df[mall_page_df["version"] == version]["create_time"].max()
            # 允许误差
            last_create_time = str(
                datetime.datetime.strptime(last_create_time, "%Y-%m-%d %H:%M:%S") - datetime.timedelta(seconds=10)
            )
            # 获取下一个版本的第一个创建时间
            next_version_first_create_time = mall_page_df[mall_page_df["version"] > version]["create_time"].min()

            if next_version_first_create_time is np.nan:
                next_version_first_create_time = "2099-01-01"

            # 筛选出在当前版本的最后一个创建时间和下一个版本的第一个创建时间之间的商品明细
            mall_detail_info_in_version = mall_detail_info_df[
                (mall_detail_info_df["create_time"] >= last_create_time)
                & (mall_detail_info_df["create_time"] < next_version_first_create_time)
            ]
            mall_detail_info_in_version["version"] = version
            result_mall_detail_info_df = pd.concat(
                [result_mall_detail_info_df, mall_detail_info_in_version], ignore_index=True
            )
        return result_mall_detail_info_df

    def fill_version_goods(self, goods_df, mall_page_df, result_goods_df):
        # 按照上面的数据版本找到在创建时间之后下一个创建时间之前，采集的商品明细也分配成对应的数据版本
        # 获取商铺的数据版本列表
        goods_df = goods_df[goods_df["mall_id"].isin(mall_page_df["mall_id"])]
        if goods_df.empty:
            return result_goods_df
        versions = mall_page_df["version"].sort_values(ascending=True).unique()
        for version in versions:
            # 获取当前版本的最后一个创建时间
            last_create_time = mall_page_df[mall_page_df["version"] == version]["create_time"].max()
            # 允许误差
            last_create_time = str(
                datetime.datetime.strptime(last_create_time, "%Y-%m-%d %H:%M:%S") - datetime.timedelta(seconds=10)
            )
            # 获取下一个版本的第一个创建时间
            next_version_first_create_time = mall_page_df[mall_page_df["version"] > version]["create_time"].min()

            if next_version_first_create_time is np.nan:
                next_version_first_create_time = "2099-01-01"

            # 筛选出在当前版本的最后一个创建时间和下一个版本的第一个创建时间之间的商品明细
            # print(version, last_create_time, next_version_first_create_time)
            goods_in_version = goods_df[
                (goods_df["create_time"] >= last_create_time)
                & (goods_df["create_time"] < next_version_first_create_time)
            ]
            # 为这些商品明细分配当前版本
            goods_in_version["version"] = version
            # print(goods_in_version["create_time"])
            # 将分配了版本的商品明细合并到原始的商品明细DataFrame中
            result_goods_df = pd.concat([result_goods_df, goods_in_version], ignore_index=True)
        return result_goods_df

    def fill_na_merge(self, df_old, df_new):
        if df_new.empty:
            return df_old
        df_old = pd.merge(df_old, df_new, on=["mall_id", "version"], how="left", suffixes=("_x", "_y")).copy()
        for col in df_old.columns:
            if col.endswith("_x") and col[:-2] + "_y" in df_old.columns:
                df_old[col[:-2]] = df_old[col].fillna(df_old[col[:-2] + "_y"])
                if col[:-2] in ['shop_status']:
                    df_old[col[:-2]] = df_old[col[:-2]].mask(df_old[col] == '该店铺已失效', '该店铺已失效')
                    df_old[col[:-2]] = df_old[col[:-2]].mask(df_old[col[:-2] + "_y"] == '该店铺已失效', '该店铺已失效')
                    df_old[col[:-2]] = df_old[col[:-2]].mask(df_old[col] == '该店铺正在上传商品', '该店铺正在上传商品')
                    df_old[col[:-2]] = df_old[col[:-2]].mask(df_old[col[:-2] + "_y"] == '该店铺正在上传商品', '该店铺正在上传商品')
                if col[:-2] in ['flag']:
                    df_old[col[:-2]] = df_old[col[:-2]].mask(df_old[col] == -1, -1)
                    df_old[col[:-2]] = df_old[col[:-2]].mask(df_old[col[:-2] + "_y"] == -1, -1)
                df_old.drop([col, col[:-2] + "_y"], axis=1, inplace=True)
        return df_old

    def run_core(self, source_df):
        # 设置店铺的数据版本
        version_df_dict = self.set_version(df_pro=source_df)

        if not version_df_dict:
            return

        mall_page_df = version_df_dict["mall_page_df"]
        goods_df = version_df_dict["goods_df"]
        result_mall_detail_info_df = version_df_dict["result_mall_detail_info_df"]
        result_mall_info_df = version_df_dict["result_mall_info_df"]
        result_mall_combination_df = version_df_dict["result_mall_combination_df"]

        mall_page_df["create_time"] = mall_page_df["new_create_time"]
        mall_page_df.drop(columns=["new_create_time"], inplace=True)

        # 解析首页
        mall_rows = mall_page_df.groupby(["mall_id", "version"], as_index=False).apply(
            lambda df_one: self._parse_mall_page(df_one)
        )

        if "shop_name" not in mall_rows:
            mall_rows["shop_name"] = None

        name_mall_id_mapping = dict(zip(mall_rows["mall_id"], mall_rows["shop_name"].replace({np.nan: None})))

        if not goods_df.empty:
            goods_rows_df = (
                goods_df.groupby(["mall_id", "version"], as_index=False)
                .apply(lambda df_one: self._parse_goods(df_one))
                .replace({np.nan: None})
            )
            if not goods_rows_df.empty:
                goods_rows_df["mall_name"] = goods_rows_df["mall_id"].map(lambda x: name_mall_id_mapping.get(str(x)))
                # 根据销售额计算排名
                goods_rows_df["goods_rank"] = goods_rows_df.groupby(["mall_id", "version"])[
                    "sales_money_calculate"
                ].rank(ascending=False, method="first", na_option="bottom")
                goods_row_crawler_goods_num_df = (
                    goods_rows_df.groupby(["mall_id", "version"])
                    .agg(
                        calculate_sales=("sale_num_by_tips", "sum"),
                        calculate_sales_money=("sales_money_calculate", "sum"),
                        crawler_goods_num=("goods_id", "count"),  # 统计该组商品数量
                    )
                    .reset_index()
                )
                if not goods_row_crawler_goods_num_df.empty:
                    goods_row_crawler_goods_num_df['calculate_sales'] = goods_row_crawler_goods_num_df['calculate_sales'].map(lambda x: None if not x else x)
                    goods_row_crawler_goods_num_df['calculate_sales_money'] = goods_row_crawler_goods_num_df['calculate_sales_money'].map(lambda x: None if not x else x)

                mall_rows = self.fill_na_merge(mall_rows, goods_row_crawler_goods_num_df)
                goods_rows_datas = goods_rows_df.to_dict("records")
                self.local_db.add_batch_smart(
                    self._table_source_pdd_goods_detail,
                    goods_rows_datas,
                    update_columns=list(goods_rows_datas[0].keys()),
                )

        if not result_mall_detail_info_df.empty:
            result_mall_detail_info_row_df = (
                result_mall_detail_info_df.groupby(["mall_id", "version"], as_index=False)
                .apply(lambda df_one: self._parse_one_mall_detail(df_one))
                .replace({np.nan: None})
            )
            mall_rows = self.fill_na_merge(mall_rows, result_mall_detail_info_row_df)

        if not result_mall_info_df.empty:
            result_mall_info_row_df = (
                result_mall_info_df.groupby(["mall_id", "version"], as_index=False)
                .apply(lambda df_one: self._parse_mall_info(df_one))
                .replace({np.nan: None})
            )
            mall_rows = self.fill_na_merge(mall_rows, result_mall_info_row_df)

        if not result_mall_combination_df.empty:
            result_mall_combination_row_df = (
                result_mall_combination_df.groupby(["mall_id", "version"], as_index=False)
                .apply(lambda df_one: self._parse_one_mall_combination(df_one))
                .replace({np.nan: None})
            )
            mall_rows = self.fill_na_merge(mall_rows, result_mall_combination_row_df)

        mall_rows.replace({np.nan: None}, inplace=True)
        mall_rows["shop_name"] = mall_rows["shop_name"].fillna("")
        mall_rows_datas = mall_rows.to_dict("records")
        self.local_db.add_batch_smart(
            self._table_source_pdd_shop_info, mall_rows_datas, update_columns=list(mall_rows_datas[0].keys())
        )

    @retry(stop_max_attempt_number=5, wait_fixed=10000)
    def find_sql(self, sql, **kwargs):
        return self.to_db.find(sql, **kwargs)

    @logger.catch
    def run_by_mall_ids(self, mall_ids):
        try:
            batch_size = 100
            for idx in range(0, len(mall_ids), batch_size):
                mall_ids_str = ",".join([f"'{i}'" for i in mall_ids[idx:idx + batch_size]])
                datas = self.find_sql(
                    f"select mall_id, id, page_no, base_url, response, create_time from {self._table_net_pdd_proxy_log} where mall_id in ({mall_ids_str})",
                    to_json=True,
                )
                self.run_core(pd.DataFrame(datas))
        except Exception as e:
            logger.error(traceback.format_exc())

    def run(self):
        sql = f"""select mall_id from {self._table_net_pdd_shop_info}"""
        sql = f"SELECT distinct mall_id FROM {self._table_source_pdd_shop_info} WHERE shop_name = ''"
        # sql = f"""select mall_id from {self._table_net_pdd_shop_info} where last_crawler_time > '2024-12-01' and last_crawler_time <= '2025-01-01'"""
        sql = f"""select distinct mall_id from {self._table_net_pdd_proxy_log} where create_time > '2025-04-25 08:50' and create_time <= '2025-04-25 11:00'"""
        # sql = """ SELECT mall_id FROM wfq_source.net_pdd_shop_info_task WHERE channel = 'wfq_test_210'"""
        # sql = """
        # SELECT mall_id
        # FROM result_pdd_shop_info
        # WHERE version in ('2025-02-25','2025-02-26','2025-02-27','2025-02-28','2025-03-01','2025-03-02','2025-03-03','2025-03-04','2025-03-05','2025-03-06','2025-03-07','2025-03-08','2025-03-09','2025-03-10')
        # """
        # sql = """SELECT mall_id FROM source_net_pdd_shop_info where mall_id in ('798043560','798043573','798045121','798045656','798055933','798058641','798062','798062179','798063979','798066195')"""
        # datas = [i["mall_id"] for i in self.find_sql(sql, to_json=True)]
        # pd.DataFrame(datas).to_csv("cache.csv")
        datas = ["781292378"]
        batch_size = 10
        logger.info(f"总计需处理 {len(datas)} 条数据")
        tasks = []
        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            for i in range(0, len(datas), batch_size):
                tasks.append(executor.submit(self.run_by_mall_ids, datas[i:i + batch_size]))

            for task in tqdm(
                concurrent.futures.as_completed(tasks), total=len(tasks), desc=f"爬虫任务 pid:{os.getpid()}"
            ):
                if task.exception():
                    raise task.exception()


if __name__ == "__main__":
    SyncPddSource().run()
